package com.networkflow_analysis

import com.networkflow_analysis.bean.{PvCount, UserBehavior}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/29 23:26
  *网站总浏览量
  *容易产生数据倾斜
  */
object PageView {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    //读取数据，转换成样例类提取时间戳和watermark
    val resource = getClass.getResource("/").toURI
    val filePath: String = "NetworkFlowAnalysis/src/main/resources/UserBehavior.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val dataStream: DataStream[UserBehavior] = fileStream.map(data => {
      val arr = data.split(",");
      UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, arr(3), arr(4).toLong)
    })
      .assignAscendingTimestamps(_.timestamp * 1000L)

    val resultStream: DataStream[PvCount] = dataStream.filter(_.behavior == "pv")
      .map(data => ("pv", 1L)) //定义一个pv字符串作为分组的dummy key
      .keyBy(_._1) //所有数据会被分到同一个组
      .timeWindow(Time.hours(1)) //一小时滚动窗口
      .aggregate(new PvCountAgg, new PvCountWindowResult)
    resultStream.print("one group")


    env.execute("pv job")
  }

  //自定义预聚合函数
  class PvCountAgg extends AggregateFunction[(String, Long), Long, Long] {
    override def createAccumulator(): Long = 0L

    override def add(in: (String, Long), acc: Long): Long = acc + 1

    override def getResult(acc: Long): Long = acc

    override def merge(acc: Long, acc1: Long): Long = acc + acc1
  }

  //自定义窗口函数
  class PvCountWindowResult extends WindowFunction[Long, PvCount, String, TimeWindow] {
    override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[PvCount]): Unit = {
        out.collect(PvCount(window.getEnd,input.head))
    }
  }

}
